
    /**  
    * @Title: SocketIOStarterConfiguration.java
    * @Package com.efast.cafe.commponent.es.socketio.configuration
    * @Description: TODO(用一句话描述该文件做什么)
    * @author Administrator
    * @date 2018年10月29日
    * @version V1.0  
    */
    
package com.efast.cafe.commponent.es.socketio.configuration;

import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.TimeUnit;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;

import org.slf4j.MDC;
import org.slf4j.event.Level;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.GenericBeanDefinition;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.ScannedGenericBeanDefinition;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.StringRedisTemplate;

import com.efast.cafe.util.BeanDefinitionFilter;
import com.efast.cafe.util.ConfigUtil;
import com.efast.cafe.util.IpAddressUtil;
import com.efast.cafe.util.ProxyDelegate;
import com.cafeauto.socketio.SocketCallbackData;
import com.cafeauto.socketio.SocketCallbackMessage;
import com.cafeauto.socketio.SocketIOInData;
import com.cafeauto.socketio.SocketIOInMessage;
import com.corundumstudio.socketio.AckRequest;
import com.corundumstudio.socketio.SocketConfig;
import com.corundumstudio.socketio.SocketIOClient;
import com.corundumstudio.socketio.SocketIOServer;
import com.corundumstudio.socketio.Transport;
import com.corundumstudio.socketio.listener.ConnectListener;
import com.corundumstudio.socketio.listener.DataListener;
import com.corundumstudio.socketio.listener.DisconnectListener;
import com.corundumstudio.socketio.listener.ExceptionListener;
import com.efast.cafe.commponent.es.socketio.annotation.SocketDisconnectListener;
import com.efast.cafe.commponent.es.socketio.annotation.SocketIOAlarm;
import com.efast.cafe.commponent.es.socketio.annotation.SocketIOEvent;
import com.efast.cafe.commponent.es.common.entity.CommonEsConfig;
import com.efast.cafe.commponent.es.common.enums.ServiceTypeEnum;
import com.efast.cafe.commponent.es.common.service.CommonEsConfigService;
import com.efast.cafe.commponent.es.socketio.annotation.SocketConnectListener;
import com.efast.cafe.commponent.es.socketio.auth.AuthorizationListener;
import com.efast.cafe.commponent.es.socketio.listener.SocketIOConnectListener;
import com.efast.cafe.commponent.es.socketio.listener.SocketIODisconnectListener;
import com.efast.cafe.commponent.es.socketio.listener.SocketIOExceptionListener;
import com.efast.cafe.commponent.es.socketio.properties.SocketIOConfig;
import com.efast.cafe.commponent.es.socketio.util.SocketIOUtil;
import com.efast.cafe.commponent.log.ext.CafeLogger;
import com.efast.cafe.commponent.log.ext.CafeLoggerFactory;
import com.efast.cafe.framework.constant.FrameConstant;
import com.efast.cafe.framework.exception.ServiceException;
import com.efast.cafe.util.SpringUtils;
import com.efast.cafe.util.StringUtil;

import net.sf.json.JSONObject;

/**
    * @ClassName: SocketIOStarterConfiguration
    * @Description: websocket 服务启动类
    * @author xuanming.feng
    * @date 2018年10月29日
    *
    */
@ConditionalOnProperty(
        name="socketio.enable",
        havingValue="true"
)
@Configuration
@AutoConfigureAfter(SocketIOConfig.class)
public class SocketIOStarterConfiguration implements ApplicationListener<ContextRefreshedEvent>{

	CafeLogger logger = CafeLoggerFactory.getLogger(getClass(),ServiceTypeEnum.SOCKETIO.name());
	
	
	
	private boolean isReady = false;
	
	private  SocketIOServer server= null;
	@Autowired
	private SocketIOConfig socketIOConfig;
	@Autowired
	private StringRedisTemplate stringRedisTemplate;
	@Autowired
	private CommonEsConfigService commonEsConfigService;
	
	@Override
	public void onApplicationEvent(ContextRefreshedEvent event) {
		/**
		 * spring容器刷新事件触发
		 */
		try{
			ContextRefreshedEvent contextRefreshedEvent = (ContextRefreshedEvent) event;
			ApplicationContext ctx = contextRefreshedEvent.getApplicationContext();
			try {
				if(!isReady){
					startSocketServer(ctx);
					isReady = true;
				}
			} catch (Exception e) {
				logger.error("startSocketServer 异常",e);
				stopSocketServer(ctx);
			}
		}catch(Exception e){
			e.printStackTrace();
		}
		
	}
	
	private String getSingleIp(List<String> iplist){
		String[] configIps = ConfigUtil.getValue("socketio.instance").split(",");
		for(String ip:iplist){
			for(String cip : configIps){
				if(ip.equals(cip)){
					return ip;
				}
			}
		}
		return null;
	}
	
	
	public void startSocketServer(ApplicationContext ctx)throws Exception{
		String port = socketIOConfig.getPort();
		if(StringUtil.isNotBlank(port)){
			List<String> iplist = IpAddressUtil.getLocalIPS();
			String localIP = getSingleIp(iplist);
			if(null==localIP){
				logger.warn("没有找到配置对应的ip,无法启动socket io服务。");
				return;
			}
			logger.info("正在准备启动socket io服务，ip:"+localIP);
			String mac = IpAddressUtil.getMacByIp(localIP);
			stringRedisTemplate.delete(FrameConstant.SOCKET_IO_ONLINE_TOKEN+":"+localIP);
			SpringUtils springUtils = SpringUtils.getInstance(ctx);
			//存放socket io服务的事件监听类
			List<Class<?>> eventListenerList = new ArrayList<>();
			//存放socket io服务的连接监听类
			List<Class<?>> connectListenerList = new ArrayList<>();
			//存放socket io 断开连接监听类
			List<Class<?>> dicConnectList = new ArrayList<>();
			//存放设备掉线警告处理类
			List<Class<?>> dicConnectAlarmList = new ArrayList<>();
			//扫描指定包路径下 包含@SocketIOEvent注解的类
			if(null!=socketIOConfig.getEventPackages()){
				generateSocketList(springUtils, eventListenerList, connectListenerList, dicConnectList,dicConnectAlarmList);
			}
			
			//构造SocketIO服务
			com.corundumstudio.socketio.Configuration config = new com.corundumstudio.socketio.Configuration();
		//	if(StringUtil.isNotBlank(socketIOConfig.getHost()))
				//config.setHostname(socketIOConfig.getHost());
			config.setPort(Integer.parseInt(socketIOConfig.getPort()));
			config.setAuthorizationListener(new AuthorizationListener());
			config.setTransports(Transport.POLLING,Transport.WEBSOCKET);
			config.setPingTimeout(socketIOConfig.getPingTimeout());
			config.setPingInterval(socketIOConfig.getPingInterval());
			config.setBossThreads(socketIOConfig.getBossThreads());
			config.setWorkerThreads(socketIOConfig.getWorkerThreads());
			config.setExceptionListener(new SocketIOExceptionListener(socketIOConfig.isEnableExceptionLog()));
			//boolean useRedissonStore = socketIOConfig.isUseRedissonStore();
			//to do
//			SocketConfig socketConfig = new SocketConfig();
//			
//			config.setSocketConfig(socketConfig);
			server = new SocketIOServer(config);
			socketIOConfig.setInstance(localIP);
			server.addConnectListener(new SocketIOConnectListener(socketIOConfig));
			server.addDisconnectListener(new SocketIODisconnectListener(socketIOConfig));
			//server.setPipelineFactory(new MySocketIOChannel());
			//增加客户端连接事件监听
			for(Class<?> c : connectListenerList){
				try{
				ConnectListener connectListener = (ConnectListener)c.newInstance();
				server.addConnectListener(connectListener);
				logger.info("socket io server("+port+")，add connect listener : "+c.getName());
				}catch(ClassCastException e){
					logger.error(c.getName()+":必须实现com.corundumstudio.socketio.listener.ConnectListener", e);
				}
			}
			for(Class<?> c : dicConnectList){
				try{
					DisconnectListener disconnectListener = (DisconnectListener)c.newInstance();
					server.addDisconnectListener(disconnectListener);
					logger.info("socket io server("+port+")，add disconnect listener : "+c.getName());
				}catch(Exception e){
					logger.error(c.getName()+":必须实现com.corundumstudio.socketio.listener.DisconnectListener", e);
				}
			}
			
			//增加事件驱动
			for(Class<?> c : eventListenerList){
				final String simpleName = c.getSimpleName();
				logger.info("SocketIO server("+port+") add event:"+simpleName);
				server.addEventListener(simpleName,String.class,new DataListener<String>() {
					@Override
					public void onData(SocketIOClient client,String dataStr,AckRequest ackRequest){
						String uid = client.get(FrameConstant.SOCKET_UID);
						MDC.put("userName", uid);
						SocketCallbackData callbackData = null;
						CommonEsConfig config =	commonEsConfigService.getConfig(ServiceTypeEnum.SOCKETIO.name(), simpleName, null);
						if(null==config){
							callbackData = new SocketCallbackData();
							String callres = callbackData.callbackResult(false, null, SocketCallbackData.ERROR_CONFIG_ERROR, null, null);
							if (ackRequest.isAckRequested()) {
								ackRequest.sendAckData(callres);
							}
							logger.error(simpleName+"; The interface config is undefined.");
							return;
						}
						
						boolean enableLog = (config.getLogs()!=null && config.getLogs().intValue()==1);
						if(enableLog){
							logger.cacheLog(Level.INFO, dataStr, "SocketIO,"+simpleName+",Received a message.");
						}
						
						
						/**
						 * dataStr 字符串标准格式 {"message":{"messageId":"xxx","busObject":""},"hmac":"","time":"2017-6-13 15:15","version":"1"}
						 * message消息对象 
						 * 	message.messageId 消息id
						 * 	message.busObject 业务数据
						 * hmac 动态令牌
						 * time 服务器时间
						 * version 数据传输协议版本号
						 * 
						 *响应 字符串标准格式:{"success":true,"errorCode":"","errorMessage","message":{"messageId":"xxx","busObject":"{}"},"hmac":"","time":"yyyy-MM-dd HH:mm"} 
						 * errorCode 错误代码
						 * errorMessage 错误消息
						 * message 同上
						 * hmac 同上
						 * time 服务器响应当前时间
						 */
						
						SocketIOInData socketIOInData = null;
						SocketIOInMessage socketInMessage = null;
						//Date sendTime = null;
						try{
							
							JSONObject dataJson = new JSONObject().fromObject(dataStr);
							socketIOInData = (SocketIOInData)JSONObject.toBean(dataJson,SocketIOInData.class);
							if(socketIOInData!=null){
								socketInMessage = socketIOInData.getMessage();
							}
							//Date sendTime = DateUtils.parse(socketIOInData.getTime(), DateUtils.FORMAT_DATE_YYYY_MM_DD_HH_MM_SS);
						}catch(Exception e){
							//解析json数据错误 ERROR_DATA_ERROR
							callbackData = new SocketCallbackData();
							String callres = callbackData.callbackResult(false, null, SocketCallbackData.ERROR_DATA_ERROR, null, null);
							if (ackRequest.isAckRequested()) {
								ackRequest.sendAckData(callres);
							}
							if(enableLog){
								logger.cacheLog(Level.ERROR, callres, "SocketIO Event "+simpleName+",DATA_ERROR");
							}
							logger.error(simpleName+";request message;Parse JSON error:", e);
							return;
						}
						if(null!=socketIOInData && socketInMessage!=null){
							//check data:messageId
							if(StringUtil.isEmpty(socketInMessage.getMessageId()) || StringUtil.isEmpty(socketIOInData.getVersion())){
								callbackData = new SocketCallbackData();
								SocketCallbackMessage cmessage = new SocketCallbackMessage();
								cmessage.setBusObject(socketInMessage.getBusObject());
								cmessage.setMessageId(socketInMessage.getMessageId());
								String callres = callbackData.callbackResult(false, null, SocketCallbackData.ERROR_DATA_ERROR, null, cmessage);
								if (ackRequest.isAckRequested()) {
									ackRequest.sendAckData(callres);
								}
								if(enableLog){
									logger.cacheLog(Level.ERROR, callres, "SocketIO Event "+simpleName+",DATA_ERROR");
								}
								logger.error(simpleName+";request message;messageId or version is null");
								return;
							}
							
							try{
								String ver = socketIOInData.getVersion();
								String procClass = c.getName();
								if(StringUtil.isNotBlank(ver) && !ver.equals("1")){
									procClass = procClass+(ver.replaceAll("\\.", "_"));
								}
								callbackData = (SocketCallbackData)ProxyDelegate.invoke(procClass, "onMessage", new Object[]{client,socketIOInData});
								String callres = callbackData.callbackResult(callbackData.isSuccess(),
										callbackData.getHmac(), callbackData.getErrorCode(),
										callbackData.getErrorMessage(), callbackData.getMessage());
								
								if (ackRequest.isAckRequested()) {//客户端存在回调函数
									ackRequest.sendAckData(callres);//发送响应到客户端
									if(enableLog){
										logger.cacheLog(Level.INFO, callres, "SocketIO Event "+simpleName+",Response success.");
									}
								}else{
									//客户端不存在回调函数 记录日志
									logger.warn("客户端请求，没有回调服务端响应函数,无法接收服务端响应.");
									if(enableLog){
										logger.cacheLog(Level.INFO, callres, "SocketIO Event "+simpleName+",Response error,client no definition callback.");
									}
								}
							}catch(Exception e){
								callbackData = new SocketCallbackData();
								if(e instanceof InvocationTargetException){
									InvocationTargetException ine=(InvocationTargetException)e;
									Throwable targetException=ine.getTargetException();
									if(targetException instanceof ServiceException){
										logger.error("业务异常", e);
										callbackData.setErrorCode(targetException.getMessage());
									}else if(targetException instanceof ClassNotFoundException){
										callbackData.setErrorCode(SocketCallbackData.ERROR_VERSION_ERROR);
									}else{
										logger.error("系统异常", e);
										callbackData.setErrorCode(SocketCallbackData.ERROR_SYSTEM_ERROR);
									}
								}else{
									logger.error("系统异常", e);
									callbackData.setErrorCode(SocketCallbackData.ERROR_SYSTEM_ERROR);
								}
								SocketCallbackMessage cmessage = new SocketCallbackMessage();
								cmessage.setBusObject(socketInMessage.getBusObject());
								cmessage.setMessageId(socketInMessage.getMessageId());
								String callres = callbackData.callbackResult(false, "", callbackData.getErrorCode(), null, cmessage);
								if (ackRequest.isAckRequested()) {
									ackRequest.sendAckData(callres);//响应异常信息到客户端
									//记录日志
									if(enableLog){
										logger.cacheLog(Level.ERROR, callres, "SocketIO Event "+simpleName+",exception:"+callbackData.getErrorCode());
									}
								}else{
									//客户端不存在回调函数 记录日志
									logger.warn("客户端请求，没有回调服务端响应函数,无法接收服务端响应.");
									if(enableLog){
										logger.cacheLog(Level.INFO, callres, "SocketIO Event "+simpleName+",Response error,client no definition callback(exception).");
									}
								}
							}
							
						}else{
							//数据错误 ERROR_DATA_ERROR 
							//解析json数据错误 ERROR_DATA_ERROR
							callbackData = new SocketCallbackData();
							String callres = callbackData.callbackResult(false, null, SocketCallbackData.ERROR_DATA_ERROR, null, null);
							if (ackRequest.isAckRequested()) {
								ackRequest.sendAckData(callres);
							}
							if(enableLog){
								logger.cacheLog(Level.ERROR, callres, "SocketIO Event "+simpleName+",DATA_ERROR");
							}
							return;
						}
						
						
					}
				});
			}
			
			server.startAsync().addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (future.isSuccess()) {
                        isReady = true;
                        logger.info("Socket io server started by "+port);
                        HashOperations<String,String,String> ops = stringRedisTemplate.opsForHash();
                        if(!ops.hasKey(FrameConstant.SOCKET_IO_INSTANT, localIP)){
                        	ops.put(FrameConstant.SOCKET_IO_INSTANT, localIP, localIP);
                        }
                        Timer timer = new Timer();
                        timer.schedule(new TimerTask() {
							@Override
							public void run() {
								//logger.info("SOCKET_IO_ONLINE_TOKEN set expire...");
								if(stringRedisTemplate.hasKey(FrameConstant.SOCKET_IO_ONLINE_TOKEN+":"+localIP)){
									stringRedisTemplate.boundHashOps(FrameConstant.SOCKET_IO_ONLINE_TOKEN+":"+localIP)
									.expire(120, TimeUnit.SECONDS);
								}
							}
						},0,60000);
                        
                        Timer timer1 = new Timer();
                        timer1.schedule(new TimerTask() {
							@Override
							public void run() {
								HashMap<String,Integer> disconnectCounterMap = SocketIOUtil.countDisconnectCounterByUid(socketIOConfig.getAlarm().getSingleDisconnectTimer());
								Integer totalCounter = SocketIOUtil.countTotalDisconnectCounter();
								SocketIOUtil.clearAllDisconnectCounter();
								if(!disconnectCounterMap.isEmpty() && dicConnectAlarmList.size()>0){
									try {
										ProxyDelegate.invoke(dicConnectAlarmList.get(0).getName(), "disconnectAlarm", new Object[]{localIP,disconnectCounterMap});
									} catch (Exception e) {
										logger.error("执行掉线告警处理类失败", e);
									}
								}
								if(totalCounter!=null && totalCounter.intValue()>=socketIOConfig.getAlarm().getAllDisconnectTimer()){
									try {
										ProxyDelegate.invoke(dicConnectAlarmList.get(0).getName(), "disconnectAllAlarm", new Object[]{localIP,totalCounter.intValue()});
									} catch (Exception e) {
										logger.error("执行掉线告警处理类失败", e);
									}
								}
								
							}
						},0,socketIOConfig.getAlarm().getDisconnectFrequency());
                        
                    } else {
                        logger.error("长连接端口:{},启动失败!!!!!!!!!!!!!!!!!!!!!!!!", port, future.cause());
                    }
                }
            });
			
			
		}else{
			logger.warn("The service failed to start, and the configuration parameters were indeed configured(port).");
		}
		
	}


	
	    /**
	    * @Title: generateSocketList
	    * @Description: 扫描返回event 、connect、disconnect的类
	    * @param @param springUtils
	    * @param @param eventListenerList
	    * @param @param connectListenerList
	    * @param @param dicConnectList    参数
	    * @return void    返回类型
	    * @throws
	    */
	private void generateSocketList(SpringUtils springUtils, List<Class<?>> eventListenerList,
			List<Class<?>> connectListenerList, List<Class<?>> dicConnectList, List<Class<?>> dicConnectAlarmList) {
		socketIOConfig.getEventPackages().forEach(packageStr->{
			Set<BeanDefinition> candidateComponents = springUtils.findCandidateComponents(packageStr,
					new BeanDefinitionFilter() {
						public boolean filter(ScannedGenericBeanDefinition sbd) {
							try {
								GenericBeanDefinition definition = sbd;
								Class<?> entityClazz = Class.forName(definition.getBeanClassName());
								SocketIOEvent annotation = (SocketIOEvent) entityClazz
										.getAnnotation(SocketIOEvent.class);
								SocketConnectListener listener = (SocketConnectListener)entityClazz.getAnnotation(SocketConnectListener.class);
								SocketDisconnectListener dicconnectListener = (SocketDisconnectListener)entityClazz.getAnnotation(SocketDisconnectListener.class);  
								SocketIOAlarm alarm = (SocketIOAlarm)entityClazz.getAnnotation(SocketIOAlarm.class);
								if (annotation == null && listener==null && dicconnectListener==null && alarm==null) {
									return false;
								}
								return true;
							} catch (ClassNotFoundException e) {
								e.printStackTrace();
							}
							return false;
						}
					});
			
			for (BeanDefinition beanDefinition : candidateComponents) {
				GenericBeanDefinition definition = (GenericBeanDefinition) beanDefinition;
				try {
					Class<?> interfaceClazz = Class.forName(definition.getBeanClassName());
					SocketIOEvent annotation =  interfaceClazz.getAnnotation(SocketIOEvent.class);
					SocketConnectListener listenerAnnotation = null;
					SocketDisconnectListener socketDisconnectListenerAnnotation = null;
					SocketIOAlarm alarm = null;
					if(null==annotation){
						listenerAnnotation = interfaceClazz.getAnnotation(SocketConnectListener.class);
					}
					if(null==listenerAnnotation){
						socketDisconnectListenerAnnotation = interfaceClazz.getAnnotation(SocketDisconnectListener.class);
					}
					if(null==socketDisconnectListenerAnnotation){
						alarm = interfaceClazz.getAnnotation(SocketIOAlarm.class);
					}
					
					if(null!=annotation){
						eventListenerList.add(interfaceClazz);
					}else if(null!=listenerAnnotation){
						connectListenerList.add(interfaceClazz);
					}else if(null!=socketDisconnectListenerAnnotation){
						dicConnectList.add(interfaceClazz);
					}else if(alarm!=null){
						dicConnectAlarmList.add(interfaceClazz);
					}
					
				} catch (ClassNotFoundException e) {
					e.printStackTrace();
				}
			}
			
		});
	}
	
	
	public void stopSocketServer(ApplicationContext ctx){
		if(server!=null){
			server.getAllClients().forEach(client->{
				client.disconnect();
			});
			server.stop();
		}
	}

}
